package br.unifor.g2cl;

import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.sf.jgcs.NotJoinedException;
import net.sf.jgcs.Service;
import net.sf.jgcs.membership.MembershipListener;
import net.sf.jgcs.membership.MembershipSession;

/**
 * ReplicatedHashMap
 * 
 * @author Garotinho
 */
public class ReplicatedHashMap<K extends Serializable, V extends Serializable>
		implements Map<K, V>, StateListener, MembershipListener {

	private static Logger logger = Logger.getLogger(ReplicatedHashMap.class.getName());
	
	private Map<K, V> map;

	// isso aqui é pra ele só replicar se tiver mais de 1 no grupo
	private volatile boolean isAlone;
	private RpcDispatcher rpc;
	private boolean hasState;
	private Object lock_state = new Object();
	
	private MembershipSession membershipSession;

	
	
	public ReplicatedHashMap(IMarshalDataSession marshalDataSession,
			MembershipSession membershipSession, Service service)
			throws IOException {
		this.hasState = false;
		this.isAlone = false;
		this.membershipSession = membershipSession;
		membershipSession.setMembershipListener(this);
		
		StateTransferDataSession state = new StateTransferDataSession(marshalDataSession, service,
				membershipSession);
		state.setStateListener(this);
		
		if (membershipSession.getMembership().getMembershipList().size() == 1) {
			isAlone = true;
			hasState = true;
		} else {
			logger.log(Level.INFO, "I'm not the first member. Getting existing hash map");
			state.requestState();
			synchronized (lock_state) {
				try {
					while (!hasState) {
						lock_state.wait();
					}
				} catch (InterruptedException e) {
					// FIXME lançar exceção pro chamador, abortando a criação do objeto???
					e.printStackTrace();
				}
			}
		}

		this.rpc = new RpcDispatcher(marshalDataSession, membershipSession,
				service, this);
		rpc.setLocal(false);
	}

	public boolean containsKey(Object key) {
		return map.containsKey(key);
	}

	public boolean containsValue(Object value) {
		return map.containsValue(value);
	}

	public Set<java.util.Map.Entry<K, V>> entrySet() {
		return map.entrySet();
	}

	public V get(Object key) {
		return map.get(key);
	}

	public boolean isEmpty() {
		return map.isEmpty();
	}

	public Set<K> keySet() {
		return map.keySet();
	}

	public int size() {
		return map.size();
	}

	public Collection<V> values() {
		return map.values();
	}

	// ================================================

	public V put(K key, V value) {
		if (!isAlone) {
			try {
				rpc.callRemoteMethod("_put", new Class[] { key.getClass(),
						value.getClass() }, new Object[] { key, value },
						GroupRequest.GET_ALL, 0);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		V v = map.put(key, value);
		return v;
	}

	public void putAll(Map<? extends K, ? extends V> m) {
		if (!isAlone) {
			try {
				rpc.callRemoteMethod("_putAll", new Class[] { m.getClass() },
						new Object[] { m }, GroupRequest.GET_ALL, 0);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		map.putAll(m);

	}

	public void clear() {
		if (!isAlone) {
			try {
				rpc.callRemoteMethod("_clear", null, null, GroupRequest.GET_ALL, 0);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		map.clear();
	}

	public V remove(Object key) {
		if (!isAlone) {
			try {
				rpc.callRemoteMethod("_remove", new Class[] { key.getClass() },
						new Object[] { key }, GroupRequest.GET_ALL, 0);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return map.remove(key);
	}

	// =================== RPC ===========================

	public void _put(K key, V value) {
		map.put(key, value);
	}

	public void _putAll(Map<? extends K, ? extends V> m) {
		map.putAll(m);
	}

	public void _clear() {
		map.clear();
	}

	public void _remove(Object key) {
		map.remove(key);
	}


	/* ********************************************************************** *
	 * ************************* State transfer related ********************* *
	 * ********************************************************************** */
	
	@SuppressWarnings("unchecked")
	public void stateIncoming(byte[] state) {
		try {
			this.map = (Map<K, V>) Util.getObjectFromByte(state, 0, state.length);
		} catch (Exception e) {
			// TODO colocar uma mensagem assustadora
			e.printStackTrace();
		}
		synchronized (lock_state) {
			hasState = true;
			lock_state.notifyAll();
		}
	}

	public byte[] stateOutgoing() {
		logger.log(Level.INFO, "state outgoing");
		try {
			return Util.getArrayFromObject(map);
		} catch (Exception e) {
			// TODO colocar uma mensagem assustadora
			e.printStackTrace();
		}
		return null;
	}

	public boolean hasState() {
		return hasState;
	}

	//////////////////////////////////
	
	
	public void onExcluded() {
		// TODO codigo para tentar se recuperar ????		
	}

	public void onMembershipChange() {
		try {
			// XXX se a gente for mudar o jgcs, sugiro acrescentar um m�todo s� pra pegar o TAMANHO da lista.
			isAlone = membershipSession.getMembership().getMembershipList().size() == 1;
		} catch (NotJoinedException e) {
			logger.log(Level.SEVERE, "Not joined???");
		}
		
	}

}
